package org.apache.flink.runtime.dispatcher.runner;

import org.apache.flink.runtime.dispatcher.DefaultDispatcherBootstrap;
import org.apache.flink.runtime.dispatcher.Dispatcher;
import org.apache.flink.runtime.dispatcher.DispatcherBootstrap;
import org.apache.flink.runtime.dispatcher.DispatcherFactory;
import org.apache.flink.runtime.dispatcher.DispatcherId;
import org.apache.flink.runtime.dispatcher.PartialDispatcherServices;
import org.apache.flink.runtime.dispatcher.PartialDispatcherServicesWithJobGraphStore;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobmanager.JobGraphWriter;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.util.FlinkRuntimeException;

import java.util.Collection;

/**
 * Factory for the {@link DefaultDispatcherGatewayService}.
 */
class DefaultDispatcherGatewayServiceFactory implements AbstractDispatcherLeaderProcess.DispatcherGatewayServiceFactory {

	private final DispatcherFactory dispatcherFactory;

	private final RpcService rpcService;

	private final PartialDispatcherServices partialDispatcherServices;

	DefaultDispatcherGatewayServiceFactory(DispatcherFactory dispatcherFactory, RpcService rpcService,
		PartialDispatcherServices partialDispatcherServices) {
		this.dispatcherFactory = dispatcherFactory;
		this.rpcService = rpcService;
		this.partialDispatcherServices = partialDispatcherServices;
	}

	@Override
	public AbstractDispatcherLeaderProcess.DispatcherGatewayService create(DispatcherId fencingToken, Collection<JobGraph> recoveredJobs,
		JobGraphWriter jobGraphWriter) {
		//  Dispatcher 的一个默认引导程序
		//  待恢复执行的 job 的集合
		final DispatcherBootstrap bootstrap = new DefaultDispatcherBootstrap(recoveredJobs);

		final Dispatcher dispatcher;
		try {

			// 创建 Dispatcher
			// dispatcherFactory = SessionDispatcherFactory
			dispatcher = dispatcherFactory.createDispatcher(rpcService, fencingToken, bootstrap, //
				//  PartialDispatcherServicesWithJobGraphStore
				PartialDispatcherServicesWithJobGraphStore.from(partialDispatcherServices, jobGraphWriter));

		} catch(Exception e) {
			throw new FlinkRuntimeException("Could not create the Dispatcher rpc endpoint.", e);
		}

		// Dispatcher 也是一个 RpcEndpoint 启动起来了之后，给自己发送一个 Hello 消息证明启动
		dispatcher.start(); //

		//  返回一个返回值
		return DefaultDispatcherGatewayService.from(dispatcher);
	}
}
